package com.li.flink.datastream

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object TransformationOp2 {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //连接socket
    val text1 = env.socketTextStream("bigdata04", 9001)
    val text2 = env.socketTextStream("bigdata04", 9002)
    //处理数据
    //    text1.union(text2).print()
    //connect 可以共享状态
    val connectStream = text1.connect(text2)
    val coMapFunction = new CoMapFunction[String, String, String] {
      //两个流各自处理各自逻辑
      override def map1(in1: String): String = in1 + 1

      override def map2(in2: String): String = in2 + 22
    }
    connectStream.map(coMapFunction).print()
    //执行
    env.execute("TransformationOp2")
  }
}
